package onlinepre

import java.sql.{Connection, DriverManager, PreparedStatement}
import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.ml.PipelineModel
import org.apache.spark.sql._
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object TestStreamingModel {
  def main(args: Array[String]): Unit = {
    // 准备环境
    val spark: SparkSession = SparkSession.builder()
      .appName("TestSparkStreamingKafka")
      .master("local[*]")
      .getOrCreate()

    // 设置日志级别
    spark.sparkContext.setLogLevel("WARN")

    //  导入隐式转换
    import spark.implicits._

    // 读取数据流中的数据
    val kafkaDatas: DataFrame = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "dw-Huake05:9092,dw-Huake06:9092,dw-Huake07:9092") // 设置kafka集群
      .option("subscribe", "test01") // 设置需要读取的主题topic
      .load() // 加载数据

    // kafkaDatas  内部数据是kafka数据(key,value)
    val kafkaDataString: Dataset[(String, String)] = kafkaDatas.selectExpr("CAST(key AS string)", "CAST(value AS string)").as[(String, String)]

    // 处理，将数据按照空格切分
    val data: Dataset[String] = kafkaDataString.flatMap(x => x._2.split(","))

    //读入离线模型
    //val ModelPath = "model\\spark-lr-model"
    val ModelPath = "hdfs:///test/model"

    val pipeLine = PipelineModel.load(ModelPath)
    //data.show()
    //配置mysql
    //val intoMYSQL: intoMysql = new intoMysql("jdbc:mysql://dw-Huake05:3306/testspark?characterEncoding=UTF-8", "root", "huake@2021")

    //val predict = pipeLine.transform(data)
    //predict.show()

    if (!data.isEmpty) {
      // 利用DSL语句对数据进行wordcount
      val dfdata: DataFrame = data.toDF("label_y", "feature_x1", "feature_x2", "feature_x3", "feature_x4", "feature_x5")
      val predict = pipeLine.transform(dfdata)

      predict.writeStream.format("console")
        .outputMode("complete")  // 输出模式
        .start() // 开启任务
        .awaitTermination() // 等待程序停止
    }
      // 输出
      data.writeStream.format("console")
        .outputMode("complete")  // 输出模式
      .start() // 开启任务
      .awaitTermination() // 等待程序停止


  }

}

